package com.lazy.mongodemo;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.bson.Document;
import org.bson.types.ObjectId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientURI;
import com.mongodb.MongoCredential;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;

public class DemoInsertMt {
    private static final Logger logger = LoggerFactory.getLogger(DemoInsertMt.class);

    private MongoClient client;
    private MongoCollection<Document> coll;

    public DemoInsertMt(String host, int port){
        List<MongoCredential> credentials = new ArrayList<MongoCredential>();
        // credentials.add(MongoCredential.createCredential("sms_test", "ecodeSms", "sms_test".toCharArray()));
        //client = new MongoClient(new ServerAddress("127.0.0.1", 27017), credentials);
        client = new MongoClient(host, port);
 
        MongoDatabase db = client.getDatabase("ecodeSms");
        coll = db.getCollection("people");
        coll.drop();
    }

//    static String json = "{ \"_id\": ObjectId(\"5656dabade19b41250b1a55a\"),    \"SEQID\": 213, \"HOSTID\": 30003311,   \"HOSTNO\": \"沪AMC601\",    \"POS_TIME\": \"2015-04-09 17:24:00.0\",    \"POS_DATAFMT\": 0, \"POS_LAT\": 30.924379, \"POS_LONG\": 121.310533,   \"POS_SPEED\": 100, \"POS_ANGLE\": 100, \"POS_STARS\": -1,  \"POS_SIGLEVEL\": \"00\",   \"GATENO\": 1,  \"REC_TIME\": \"2015-04-09 16:28:34.0\",    \"SAVEDTOHISREC\": 0,   \"POSDESCRIPTION\": \"*\",  \"ACCSTATUS\": -1,  \"GPSCHANNALTYPE\": 1,  \"HIGHBEAMSTATUS\": -1, \"LOWBEAMSTATUS\": -1,  \"LEFTTURNSTATUS\": -1, \"RIGHTTURNSTATUS\": -1,    \"BRAKESTATUS\": -1}";
    static String json = "{ \"SEQID\": 213, \"HOSTID\": 30003311,   \"HOSTNO\": \"沪AMC601\",    \"POS_TIME\": \"2015-04-09 17:24:00.0\",    \"POS_DATAFMT\": 0, \"POS_LAT\": 30.924379, \"POS_LONG\": 121.310533,   \"POS_SPEED\": 100, \"POS_ANGLE\": 100, \"POS_STARS\": -1,  \"POS_SIGLEVEL\": \"00\",   \"GATENO\": 1,  \"REC_TIME\": \"2015-04-09 16:28:34.0\",    \"SAVEDTOHISREC\": 0,   \"POSDESCRIPTION\": \"*\",  \"ACCSTATUS\": -1,  \"GPSCHANNALTYPE\": 1,  \"HIGHBEAMSTATUS\": -1, \"LOWBEAMSTATUS\": -1,  \"LEFTTURNSTATUS\": -1, \"RIGHTTURNSTATUS\": -1,    \"BRAKESTATUS\": -1}";

    public void testInsert(ExecutorService executorService) {
        long startTime = System.currentTimeMillis();
        //logger.info("startTime  {}  ", startTime);
        
        int batchNum = 1000;
        CountDownLatch latch = new CountDownLatch(batchNum);
        for (int i = 0; i < batchNum; i++) {
            executorService.submit(new Worker(this, latch));
        }
        
        try{
            latch.await();
        }catch(InterruptedException e){
            e.printStackTrace();
        }
        //shutdown() 方法在终止前允许执行以前提交的任务
        //第一阶段调用 shutdown 拒绝传入任务，然后调用 shutdownNow（如有必要）取消所有遗留的任务
        //提交的任务运行结束后关闭线程池
        // executorService.shutdown(); 
        long endTime = System.currentTimeMillis();
        //logger.info("endTime  {}  ", endTime);
        logger.info("insert 1M takes time  {}  ", (endTime - startTime));
    }
    
    private void insertBatch(int batchLimit) {
        int c = 1;
        List<Document> list = new ArrayList<Document>();
        for (int i = 0; i < batchLimit; i++) {
            Document d = Document.parse(json);
            //d.put("_id", ObjectId.get());
            d.put("idx", c);
            list.add(d);
        }
        coll.insertMany(list);
    }


    public static void main(String[] args){
        String host = "127.0.0.1";
        int port = 27017;
        int count = 10;
        if (args.length == 3) {
            host = args[0];
            port = Integer.valueOf(args[1]);
            count = Integer.valueOf(args[2]);
        }
        logger.info("connecting to mongodb {}:{}  group count {} ", host, port, count);
        logger.info("json = {}", json);
        final DemoInsertMt demo = new DemoInsertMt(host, port);
        ExecutorService executorService = Executors.newFixedThreadPool(8);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < count; i++) {
            demo.testInsert(executorService);
        }
        
        executorService.shutdown();
        long endTime = System.currentTimeMillis();
        logger.info("all take time {} ms", (endTime - startTime));
    }
    
    
    static class Worker implements Runnable { 
        CountDownLatch latch;
        DemoInsertMt demoInsert;
        
        public Worker(DemoInsertMt demoInsert, CountDownLatch latch){
            this.demoInsert = demoInsert;
            this.latch = latch;
        }
        
        public void run() { 
            try{
                demoInsert.insertBatch(1000);
            }catch(Exception e){
                e.printStackTrace();
            }finally{
                latch.countDown();
            }

        }
    }

}
